package com.study.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 *
 * @author stephen
 * @date 2019-09-25 15:07
 */
class TransformWordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("TransformWordCount")
    // 指定采集周期
    val ssc = new StreamingContext(conf, Seconds(10))
    // 创建SocketInputDStream
    val lines = ssc.socketTextStream("localhost", 9999)
    // 窗口的大小为采集周期的整数倍，窗口的步长也为采集周期的整数倍
    lines.window(Seconds(60),Seconds(20))
    // 进行统计
    // 在Driver端执行，只会执行一次
//    val result = lines.flatMap(_.split(" "))
//      .transform{
//      case rdd=>{
//        // 在Driver端执行，每个采集周期均会执行一次
//        val r = rdd.map{
//          case x=>{
//            // 在Executor端执行
//            (x,1)
//          }
//        }
//        r
//      }
//    }.reduceByKey(_+_)
//    // 打印
//    result.print()

    // 启动
    ssc.start()
    ssc.awaitTermination()
  }
}
